package com.atguigu.flink.state;

import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/25
 *
 *      广播状态适用以下场景:
 *              数据流:   处理的数据，读取配置。
 *              配置流:   配置项，配置文件。
 *                       经常更新，希望更新后的配置项，通过配置流发出去，希望所有处理数据流的算子，都可以收到配置的更新。
 *                        用更新后的配置来处理。
 *
 *        套路:
 *              ①数据流
 *              ②配置流
 *              ③把配置流设置为广播流
 *              ④ 数据流.connect(广播流)
 *              ⑤ 在后续的处理中
 *                      广播流，负责更新配置，更新后的配置自动广播到所有的数据流
 *                      数据流，负责读取配置，使用
 *
 */
public class Demo3_BroadCastState
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //①数据流
        SingleOutputStreamOperator<WaterSensor> dataDS = env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            });

        //②配置流
        SingleOutputStreamOperator<MyConf> configDS = env
            .socketTextStream("hadoop103", 8889)
            .map(new MapFunction<String, MyConf>()
            {
                @Override
                public MyConf map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new MyConf(
                        data[0],
                        data[1]
                    );
                }
            });

         //③把配置流设置为广播流
        MapStateDescriptor<String,MyConf> mapStateDescriptor = new MapStateDescriptor("config", String.class, MyConf.class);
        BroadcastStream<MyConf> broadcastStream = configDS.broadcast(mapStateDescriptor);


        //④ 数据流.connect(广播流)
        dataDS.connect(broadcastStream)
              .process(new BroadcastProcessFunction<WaterSensor, MyConf, WaterSensor>()
              {
                  //处理数据流
                  @Override
                  public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<WaterSensor> out) throws Exception {
                        //读取状态中最新的配置项
                      ReadOnlyBroadcastState<String,MyConf> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                      MyConf myConf = broadcastState.get(value.getId());

                      //使用配置
                      value.setId(value.getId() + "目前使用了:" + myConf.getConf());

                      out.collect(value);

                  }

                  //处理配置流
                  @Override
                  public void processBroadcastElement(MyConf value, Context ctx, Collector<WaterSensor> out) throws Exception {
                        //在配置流中的配置项更新时，存入状态
                      BroadcastState broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                      //更新配置信息
                      broadcastState.put(value.getId(),value);
                  }
              })
              .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }

    //配置对象
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    private static class MyConf{
        private String id;
        private String conf;
    }
}
